<?php
namespace App\Common\Lib\Com;

use App\Common as com;
use App\Common\Lib\Com\RabbitMQ as nameRabbitMQ;
use Hyperf\Amqp\AMQPConnection;
use Hyperf\Contract\ConfigInterface;
use PhpAmqpLib\Exchange\AMQPExchangeType;

/**
 * @property com\Cla\ClaBase\DataStore $dataStore
 */
class RabbitMQ extends com\Cla\ClaBase\CommonOfId\Main {

	/**
	 * @var self[]
	 */
	private static array $newPool = [];

	/**
	 * pool连接
	 * @var \Hyperf\Amqp\Connection
	 */
	// private \Hyperf\Amqp\Connection $poolContainer;

	/**
	 * @var AMQPConnection
	 */
	private AMQPConnection $AMQPConnection;

	private array $defaultConfig = [
		'host' => '127.0.0.1',
		'port' => 5672,
		'user' => '',
		'password' => '',
		'vhost' => '/',
		'params' => [
			'insist' => false,
			'loginMethod' => 'AMQPLAIN',
			'loginResponse' => null,
			'locale' => 'en_US',
			'connectionTimeout' => 3.0,
			'readWriteTimeout' => 3.0,
			'context' => null,
			'keepalive' => false,
			'heartbeat' => 3
		],
	];

	private array $currentConfig = [];

	private bool $isClose = false;

	/**
	 * RabbitMQ constructor.
	 * @param array $config
	 * @throws
	 */
	public function __construct( array $config = [] ) {
		//
		$iniConfig = com\Lib\Fun\Array_::keyNameToParseNames( $this->getIniConfig() ?? [], 1, false );
		$this->currentConfig = com\Lib\Fun\Array_::arrayMerge( $this->defaultConfig, $iniConfig, $config );
		//
		// $this->AMQPConnection = \Hyperf\Utils\ApplicationContext::getContainer()->get(\Hyperf\Amqp\ConnectionFactory::class)->getConnection( 'default' );
		$this->AMQPConnection = com\Lib\Hyperf\Amqp\ConnectionFactory::makeOfNew()->makeOfPool( 'default' );

		/*
		$this->AMQPConnection = new AMQPSwooleConnection(
			$this->currentConfig['host'],
			$this->currentConfig['port'],
			$this->currentConfig['user'],
			$this->currentConfig['password'],
			$this->currentConfig['vhost'],
			$this->currentConfig['params']['insist'],
			$this->currentConfig['params']['loginMethod'],
			$this->currentConfig['params']['loginResponse'],
			$this->currentConfig['params']['locale'],
			$this->currentConfig['params']['connectionTimeout'],
			$this->currentConfig['params']['readWriteTimeout'],
			$this->currentConfig['params']['context'],
			$this->currentConfig['params']['keepalive'],
			$this->currentConfig['params']['heartbeat'],
		);
		*/
/*
		// pool模式
		$poolFactory = \Hyperf\Utils\ApplicationContext::getContainer()->get( \Hyperf\Amqp\Pool\PoolFactory::class );
		$this->poolContainer = $poolFactory->getPool( 'default' )->get();
		$this->AMQPConnection = $this->poolContainer->getConnection();
*/
		//
		parent::__construct();
		//
/*		go(function(){
			$this->selfInspectionThread();
		});*/
		//
		// $this->selfInspectionThreadWait();
	}

	/**
	 * 自检线程
	 */
	public function selfInspectionThreadWait() {
		go( function(){
			\Swoole\Coroutine::sleep( 15 );
			d('selfInspectionThread');
/*			$selfInspectionExchange = $this->newChannel()->newExchange( 'sys.selfInspection' );
			$selfInspectionExchange->exchangeDeclare( AMQPExchangeType::TOPIC, false, true, false );
			$selfInspectionExchange->getChannel()->confirm_select();*/
			//
			d('selfInspectionThread->tick');
			$timerId = \Swoole\Timer::tick(1000, function() use( &$timerId ) {
				if($this->isClose){
					\Swoole\Timer::clear( $timerId );
				}
				d('selfInspection');
				// $selfInspectionExchange->publish( 'selfInspection', '$tmp' );
				// d('sys.mail.wait()');
				// $selfInspectionExchange->getChannel()->wait();

				$this->getAMQPConnection()->getIO()->check_heartbeat();
			} );
		} );

	}

	/**
	 * 获取配置文件配置
	 * @return array
	 */
	private function getIniConfig(  ):array {
		$configIniObj = \Hyperf\Utils\ApplicationContext::getContainer()->get( ConfigInterface::class );
		$key = 'amqp.default';
		if (! $configIniObj->has($key)) {
			throw new \InvalidArgumentException(sprintf('config[%s] is not exist!', $key));
		}
		return $configIniObj->get( $key );
	}

	public function close(  ) {

		d('rabbitMQ->close');
		$this->AMQPConnection->close();
		//
		$this->isClose = true;
	}

	public function release(  ) {
		// $this->poolContainer->release();
	}
	
	public function __destruct(  ) {
		d('RabbitMQ->__destruct');
		// 归还连接
		$this->release();
		//
		parent::__destruct();
	}

	/**
	 * 重连
	 * @return $this
	 */
	public function reconnect(  ):self {
		$this->getAMQPConnection()->reconnect();
		//
		return $this;
	}

	/**
	 * @return AMQPConnection
	 */
	public function getAMQPConnection(  ):AMQPConnection {
		return $this->AMQPConnection;
	}

	/**
	 * @return RabbitMQ\_Channel
	 */
	public function goChannelDefault(  ):nameRabbitMQ\_Channel {
		return $this->goPool( __FUNCTION__, function(){
			return nameRabbitMQ\_Channel::make( $this->dataStore );
		} );
	}

	/**
	 * @return RabbitMQ\_Channel
	 */
	public function newChannel(  ):nameRabbitMQ\_Channel {
		return nameRabbitMQ\_Channel::make( $this->dataStore );
	}
	
	// ===============================
	//
	//
	// 下面为静态方法
	//
	//
	// ===============================

	/**
	 * @param string $poolKey ['default'][默认]
	 * @return static
	 */
	public static function newOfOfPool( string $poolKey = 'default' ):self {
		// @TODO 当前不是pool模式
		// return self::make();

		if( array_key_exists( $poolKey, self::$newPool ) ){
			return self::$newPool[ $poolKey ];
		}
		self::$newPool[ $poolKey ] = self::make();
		//
		return self::$newPool[ $poolKey ];
	}

	/**
	 * @return static
	 */
	public static function newOfPoolOfDefault(  ):self {
		return self::newOfOfPool( 'default' );
	}

	/**
	 * @return static
	 */
	public static function newOfDefault(  ):self {
		return self::make();
	}
	
}
